/**
 * Wihidum - Efficient Broker Hierarchy for Publish Subscribe Middleware
 *
 * License      :   Apache License 2.0
 * Home Page    :   http://code.google.com/p/wihidum/
 * Mailing List :   wihidum@googlegroups.com
 */

package lk.mrt.cse.wihidum.delivery;

import lk.mrt.cse.wihidum.processor.PublicationMessage;
import lk.mrt.cse.wihidum.util.SelfReader;
import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.concurrent.*;

/**
 * Pool of threads dedicated to send publications
 */
public class ThreadPool extends Thread {

    private static final Logger logger = Logger.getLogger(ThreadPool.class);
    private static final int NO_OF_CONCURRENT_THREADS = 10;
    private static final LinkedBlockingQueue<PublicationMessage> publicationQueue = new LinkedBlockingQueue<PublicationMessage>();
    private ExecutorService executorService = Executors.newFixedThreadPool(NO_OF_CONCURRENT_THREADS);

    /**
     * Default constructor
     */
    public ThreadPool() {
        logger.debug("Thread pool created");
        this.start();
    }

    /**
     * Continuously check publication queue and send publications
     */
    public void run() {
        HashMap<String, String> brokerMap = SelfReader.read();
        while (true) {
            try {
                PublicationMessage publicationMessage = publicationQueue.take();
                executorService.execute(new PublicationThread(publicationMessage, brokerMap));
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * Add a publication to the queue
     *
     * @param publicationMessage Publication to be added
     * @return True if added successfully, otherwise false
     */
    public boolean enqueuePublication(PublicationMessage publicationMessage) {
        logger.debug("Publication enqueued");
        return publicationQueue.add(publicationMessage);
    }

    /**
     * Remove a publication from the queue
     *
     * @return True if removed successfully, Otherwise false
     */
    private PublicationMessage dequeuePublication() {
        logger.debug("Publication dequeued");
        return publicationQueue.poll();
    }
}
